package com.example.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.config.cache.util.CacheKeyGenerator;
import com.example.util.JdbcUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class CanalClient {

    // 从配置文件注入参数
    @Value("${canal.server.host}")
    private String host;

    @Value("${canal.server.port}")
    private int port;

    @Value("${canal.subscribe}")
    private String subscribe;

    @Value("${canal.destination}")
    private String destination;

    @Value("${canal.username}")
    private String username;

    @Value("${canal.password}")
    private String password;

    @Value("${canal.batch.size}")
    private int batchSize;

    @Value("${canal.empty.retry.count}")
    private int emptyRetryCount;

    // Canal 连接器
    private CanalConnector connector;

    // 线程池（用于异步监听）
    private ExecutorService executorService;

    @Autowired
    protected RedisTemplate<String, Object> redisTemplate;

    // 初始化连接并启动监听
    @PostConstruct
    public void init() {
        // 创建 Canal 连接
        connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(host, port),
                destination,
                username,
                password
        );

        // 初始化线程池（单线程即可，避免并发处理数据混乱）
        executorService = Executors.newSingleThreadExecutor(runnable -> {
            Thread thread = new Thread(runnable, "canal-listener-thread");
            thread.setDaemon(true); // 守护线程，随服务停止而终止
            return thread;
        });

        // 启动监听任务
        startListener();
    }

    // 启动监听逻辑
    private void startListener() {
        executorService.execute(() -> {
            try {
                // 连接 Canal 服务
                connector.connect();
                // 订阅所有库表（可指定具体库表，如 "demo.user,demo.order"）
                connector.subscribe(subscribe);
                // 回滚到上次未确认的位置，避免重复消费
                connector.rollback();

                int emptyCount = 0;
                // 循环监听（服务运行期间一直执行）
                while (!Thread.currentThread().isInterrupted()) {
                    // 获取数据
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();

                    if (batchId == -1 || size == 0) {
                        // 无数据时休眠并重试
                        emptyCount++;
                        if (emptyCount >= emptyRetryCount) { // 超过最大重试次数，短暂休眠
                            System.out.println("empty count over " + emptyCount + ",sleep 10s");
                            Thread.sleep(10000);
                        } else {
                            System.out.println("empty count " + emptyCount + ",sleep 1s");
                            Thread.sleep(1000);
                        }
                    } else {
                        // 有数据时重置空计数
                        emptyCount = 0;
                        boolean isProcessedSuccessfully = false;

                        try {
                            // 处理数据，只有全部成功才会确认
                            processEntries(message.getEntries());
                            isProcessedSuccessfully = true;
                        } catch (Exception e) {
                            System.err.println("处理Canal消息批次失败，将回滚: " + e.getMessage());
                            // 发生异常时回滚，确保消息会被重新消费
                            connector.rollback(batchId);
                        }

                        // 只有全部处理成功才确认消费
                        if (isProcessedSuccessfully) {
                            connector.ack(batchId);
                            System.out.println("批次 " + batchId + " 处理成功并确认");
                        }
                    }
                }
            } catch (InterruptedException e) {
                // 线程被中断（服务停止时），退出循环
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                // 其他异常，打印日志并尝试重连
                System.err.println("Canal 监听异常，将重试：" + e.getMessage());
                try {
                    Thread.sleep(5000); // 休眠后重试
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            } finally {
                // 异常退出时断开连接
                connector.disconnect();
            }
        });
    }

    // 处理 Canal 接收到的条目（核心业务逻辑）
    private void processEntries(List<CanalEntry.Entry> entries) throws Exception {
        for (CanalEntry.Entry entry : entries) {
            // 过滤事务开始和结束的条目
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            // 解析条目为 RowChange（包含数据变更信息）
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType(); // 事件类型：INSERT/UPDATE/DELETE

            // 打印基本信息（库名、表名、事件类型）
            String logInfo = String.format(
                    "binlog[%s:%s]，数据库：%s，表名：%s，操作类型：%s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType
            );
            System.out.println("=====> " + logInfo);

            // 处理每行数据
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    // 处理删除操作（获取删除前的数据）
                    handleDelete(entry,rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    // 处理新增操作（获取新增后的数据）
                    handleInsert(entry,rowData.getAfterColumnsList());
                } else {
                    // 处理更新操作（获取更新前后的数据）
                    handleUpdate(entry,rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                }
            }
        }
    }

    // 处理新增操作
    private void handleInsert(CanalEntry.Entry entry,List<CanalEntry.Column> columns) throws Exception {
        System.out.println("新增数据：");
        printColumns(columns);
        //无需删除
    }

    // 处理更新操作
    private void handleUpdate(CanalEntry.Entry entry,List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns) throws Exception {
        System.out.println("更新前：");
        printColumns(beforeColumns);
        System.out.println("更新后：");
        printColumns(afterColumns);

        String idName = JdbcUtil.getPrimaryKey(entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
        String idValue = getIdByColumns(afterColumns,idName);
        if(idValue == null) {
            throw new Exception("id不存在，请检查配置是否错误或不全。主要问题：\n" +
                    "===>entity.properties.JdbcIdMapping.DB_TABLE_PRIMARY_KEY_MAP————未配置对应库表的id名\n" +
                    "===>getIdByColumns————该表未设置id这区分各行唯一的标识");
        }
        String key = CacheKeyGenerator.generateKey(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), idName, idValue);
        Boolean deleteResult = redisTemplate.delete(key);
        if (deleteResult == null || !deleteResult) {
            throw new Exception("Redis删除失败，key: " + key);
        }
    }

    // 处理删除操作
    private void handleDelete(CanalEntry.Entry entry,List<CanalEntry.Column> columns) throws Exception {
        System.out.println("删除数据：");
        printColumns(columns);

        String idName = JdbcUtil.getPrimaryKey(entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
        String idValue = getIdByColumns(columns,idName);
        if(idValue == null) {
            throw new Exception("id不存在，请检查配置是否错误或不全。主要问题：\n" +
                    "===>entity.properties.JdbcIdMapping.DB_TABLE_PRIMARY_KEY_MAP————未配置对应库表的id名\n" +
                    "===>getIdByColumns————该表未设置id这区分各行唯一的标识");
        }
        String key = CacheKeyGenerator.generateKey(entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), idName, idValue);
        Boolean deleteResult = redisTemplate.delete(key);
        if (deleteResult == null || !deleteResult) {
            throw new Exception("Redis删除失败，key: " + key);
        }
    }

    // 打印列数据
    private void printColumns(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(
                    "字段名：" + column.getName() +
                            "，值：" + column.getValue() +
                            "，是否更新：" + column.getUpdated()
            );
        }
    }

    // 根据id名获取id值
    private String getIdByColumns(List<CanalEntry.Column> columns, String idName) {
        for (CanalEntry.Column column : columns) {
            if (column.getName().equals(idName)) {
                return column.getValue();
            }
        }
        return null;
    }

    // 服务停止时关闭资源
    @PreDestroy
    public void destroy() {
        // 中断监听线程
        if (executorService != null) {
            executorService.shutdownNow();
        }
        // 断开 Canal 连接
        if (connector != null) {
            connector.disconnect();
        }
        System.out.println("Canal 客户端已停止");
    }
}
